package com.example.mq.mqsever.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqsever.VirtualHost;

import java.net.ConnectException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: WHY
 * Date: 2023-09-06
 * Time: 14:26
 */
public class ConsumerManager {
    //持有上层的VirtualHost对象的引用,用来操作数据
    private VirtualHost parent;
    //指定一个线程池,负责去执行具体的回调任务

    private ExecutorService workPool= Executors.newFixedThreadPool(4);
    //还得写一个存放令牌的队列(一个阻塞队列)

private BlockingQueue<String> tokenQueue=new LinkedBlockingDeque<>();
//扫描线程
    private Thread scannerThread=null;

    public ConsumerManager(VirtualHost p){
        parent=p;

        //扫描线程
        scannerThread=new Thread(()->{
            while(true){
                //获取令牌
                try{
                    String queueName=tokenQueue.take();

                    //2.根据令牌,找到队列
                    MSGQueue queue=parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[ConsumerManager] 取令牌后发现,该队列名不存在 ! queueName="+queueName);
                    }
                    //3.从这个队列消费一条消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
                //把线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }


    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
     MSGQueue queue=   parent.getMemoryDataCenter().getQueue(queueName);
        if(queue==null){
            throw new MqException("[ConsumerManager] 队列不存在! queueName="+queueName);

        }
        ConsumerEnv consumerEnv=new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumer(consumerEnv);
            //如果当前队列已经有了一些消息了,需要立即就消费掉
       int n=     parent.getMemoryDataCenter().getMessageCount(queueName);
       for(int i=0;i<n;i++){
           consumeMessage(queue);
       }

        }
    }

    private void consumeMessage(MSGQueue queue) {
        //1.按照轮询的方式.找个消费者出来
       ConsumerEnv luckyDog= queue.chooseConsumer();
        //当前队列没有消费者,
        if(luckyDog==null){
            return;
        }
        //2.从队列中取出一个消息
        Message message=  parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message==null){
            //当前队列中还没有消息,不需要消费
            return;
        }
        //3.把消息带入到消费者的回调方法中,丢给线程池执行
        workPool.submit(()->{
            try{
                //1.先把消息放到待确认的集合中去,这个操作在执行回调之前执行
                parent.getMemoryDataCenter().addMessageWaitACK(queue.getName(),message);

                //2.真正执行回调操作
                luckyDog.getConsumer().handlerDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(),
                        message.getBody());
                //3.如果是自动应答,可以直接把消息删除
                //如果是手动应答,那就先不处理,交给后续消费者调用basicAck来处理
                if(luckyDog.isAutoAck()) {
                    //1删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    //2.删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitACK(queue.getName(), message.getMessageId());
                    //3.删除内存中消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());

                }

                parent.getMemoryDataCenter().removeMessageWaitACK(queue.getName(), message.getMessageId());
            }catch (Exception e){
                e.printStackTrace();
            }


            System.out.println("[ConsumerManager] 消息被成功消费! queueName="+queue.getName());
        });

        }
    }

